package cn.zh.component;

import cn.zh.entity.User;
import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.atomic.AtomicInteger;

@Component
public class MqListener {

    private AtomicInteger seq = new AtomicInteger(0);

    //    @RabbitListener(queues = "directQueue-One")
//    @RabbitListener(bindings = @QueueBinding(
//            value = @Queue(value = "directQueue-One",durable = "false"),
//            exchange = @Exchange(value = "MqSendService-One",type = "direct", durable = "false"),
//            key = "One"
//    ))
    public void receiveDirectMsg(Message data) {
        //properties内有消息队列等信息
        MessageProperties properties = data.getMessageProperties();
        //得到消息体，变为String输出
//        User o = JSON.parseObject(data.getBody(), User.class);
//        System.out.println(o);
        String str = new String(data.getBody());
        System.out.println(str + "-----：" + seq);
        seq.incrementAndGet();
    }

    //    @RabbitListener(bindings = @QueueBinding(
//            value = @Queue(value = "directQueue-One", durable = "false"),
//            exchange = @Exchange(value = "MqSendService-One", type = "direct", durable = "false"),
//            key = "One"),
//            ackMode = "MANUAL")
    public void tsJucDirectMsg(Message data, Channel channel) throws InterruptedException, IOException {
        if (seq.get() == 3) {
            int a = 1 / 0;
            Thread.sleep(2000);
            System.out.println("sleep");
            String str = new String(data.getBody());
            System.out.println(str + "-----：" + seq);
            seq.incrementAndGet();
            channel.basicAck(data.getMessageProperties().getDeliveryTag(), false);
        }
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "directQueue-One", durable = "false"),
            exchange = @Exchange(value = "MqSendService-One", type = "direct", durable = "false"),
            key = "One"),
            ackMode = "MANUAL")
    public void tsAckDirectMsg(Message data, Channel channel) throws IOException {
        try {
            if (seq.get() == 3) {
                int a = 1 / 0;
                Thread.sleep(2000);
                System.out.println("sleep");
            }
        } catch (Exception e) {
            System.out.println("失败一次");
            seq.incrementAndGet();
            return;
        }
        String str = new String(data.getBody());
        System.out.println(str + "-----：" + seq);
        seq.incrementAndGet();
        long deliveryTag = data.getMessageProperties().getDeliveryTag();
        channel.basicAck(deliveryTag, false);
    }
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "directQueue-Two", durable = "false"),
            exchange = @Exchange(value = "MqSendService-One", type = "direct", durable = "false"),
            key = "One"),
            ackMode = "MANUAL"
    )
    public void tsJucDirectMsgTwo(Message data, Channel channel) throws IOException, InterruptedException {
        String str = new String(data.getBody());
        System.out.println(str + "-----Two：" + seq);
        System.out.println();
        seq.incrementAndGet();
        System.out.println(data.getMessageProperties().getDeliveryTag());
        System.out.println(channel.getChannelNumber());
        long deliveryTag = data.getMessageProperties().getDeliveryTag();
        channel.basicAck(deliveryTag,false);
        //channel.basicNack(data.getMessageProperties().getDeliveryTag(), false,true);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "FanoutQueue-Two", durable = "false"),
            exchange = @Exchange(value = "Fanout-Ex", type = "fanout", durable = "false"),
            key = "Two"))
    public void tsFanoutMsg(Message data, Channel channel) {
        String str = new String(data.getBody());
        System.out.println(str + "-----：" + seq);
        seq.incrementAndGet();
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "topicQueue-Two", durable = "false"),
            exchange = @Exchange(value = "Topic-Ex", type = "topic", durable = "false"),
            key = "#"))
    public void tsTopicMsg(Message data, Channel channel) {
        String str = new String(data.getBody());
        System.out.println(str + "-----：" + seq);
        seq.incrementAndGet();
    }
}

